package com.lzj.pagerank;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 *
 * 第一次  a b d
 * key a
 * value b d
 * 第二次  a 0.75 b d
 * key a
 * value 0.75 b d
 *
 * @author Administrator
 */
public class PageRankMapper extends Mapper<Text, Text, Text, Text> {

    protected void map(Text key, Text value, Context context) throws IOException, InterruptedException {
        //如果是第一次读取文件，那么需要给PR设置默认值1
        int runCount = context.getConfiguration().getInt("runCount", 1);
        //以空格切分当前行，第一个空格前为KEY，其余数据为Value
        String page = key.toString();
        //定义对象Node
        PageNode node = null;
        //判断是否为第一次加载,将数据封装成一个对象{pr,子连接}
        if (runCount == 1) {
            node = PageNode.fromMR("1.0", value.toString());
        } else {
            node = PageNode.fromMR(value.toString());
        }
        //传递老的pr值和对应的页面关系  key a  value  0.75 b d
        context.write(new Text(page), new Text(node.toString()));
        //开始计算每个节点本次应得的pr值
        if (node.containsAdjacentNodes()) {
            //每个节点的pr=当前页面pr/出链的数量
            double outValue = node.getPageRank() / node.getAdjacentNodeNames().length;
            //开始写出子节点和pr值
            for (int i = 0; i < node.getAdjacentNodeNames().length; i++) {
                String outPage = node.getAdjacentNodeNames()[i];
                //页面A投给谁，谁作为key，val是票面值，票面值为：每个节点的pr
                context.write(new Text(outPage), new Text(outValue + ""));
            }
        }
    }
}